/*
 * ====================================================================
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 * ====================================================================
 *
 * This software consists of voluntary contributions made by many
 * individuals on behalf of the Apache Software Foundation.  For more
 * information on the Apache Software Foundation, please see
 * <http://www.apache.org/>.
 *
 */
package org.examples;

import java.io.ByteArrayOutputStream;
import java.util.List;
import java.util.Map;
import java.io.IOException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.hc.client5.http.async.methods.SimpleHttpRequest;
import org.apache.hc.client5.http.async.methods.SimpleRequestBuilder;
import org.apache.hc.client5.http.config.TlsConfig;
import org.apache.hc.client5.http.entity.mime.MultipartEntityBuilder;
import org.apache.hc.client5.http.entity.mime.StringBody;
import org.apache.hc.client5.http.impl.ChainElement;
import org.apache.hc.client5.http.impl.async.CloseableHttpAsyncClient;
import org.apache.hc.client5.http.impl.async.HttpAsyncClients;
import org.apache.hc.client5.http.impl.nio.PoolingAsyncClientConnectionManager;
import org.apache.hc.client5.http.impl.nio.PoolingAsyncClientConnectionManagerBuilder;
import org.apache.hc.core5.http.*;
import org.apache.hc.core5.http.nio.entity.BasicAsyncEntityProducer;
import org.apache.hc.core5.http.nio.support.BasicRequestProducer;
import org.apache.hc.core5.http.protocol.HttpContext;
import org.apache.hc.core5.http2.HttpVersionPolicy;
import org.apache.hc.core5.io.CloseMode;
import org.apache.hc.core5.reactor.IOReactorConfig;
import org.apache.hc.core5.util.Timeout;
import org.synchronoss.cloud.nio.multipart.Multipart;
import org.synchronoss.cloud.nio.multipart.MultipartContext;
import org.synchronoss.cloud.nio.multipart.NioMultipartParser;
import org.synchronoss.cloud.nio.multipart.NioMultipartParserListener;
import org.synchronoss.cloud.nio.stream.storage.StreamStorage;

import lombok.extern.slf4j.Slf4j;

import static org.examples.DebugConstants.host;
import static org.examples.DebugConstants.schema;
import static org.examples.DebugConstants.vocsPath;

/**
 * Example of asynchronous HTTP/1.1 request execution with response streaming.
 */
@Slf4j
public class VocsClient {
  static MemOnlyPartBodyStreamStorageFactory countFactory = new MemOnlyPartBodyStreamStorageFactory();

  public static void main(final String[] args) throws Exception {
    AtomicLong startMs = new AtomicLong();

    final IOReactorConfig ioReactorConfig = IOReactorConfig.custom()
        .setSoTimeout(Timeout.ofSeconds(20))
        .build();

    final PoolingAsyncClientConnectionManager cm = PoolingAsyncClientConnectionManagerBuilder.create()
        .setDefaultTlsConfig(TlsConfig.custom().setVersionPolicy(HttpVersionPolicy.FORCE_HTTP_1).build())
        .build();

    final CloseableHttpAsyncClient client = HttpAsyncClients.custom()
        .setConnectionManager(cm)
        .setIOReactorConfig(ioReactorConfig)
//        .addRequestInterceptorFirst(new HttpRequestInterceptor() {
//
//          @Override
//          public void process(
//              final HttpRequest request,
//              final EntityDetails entity,
//              final HttpContext context) throws HttpException, IOException {
//
//            log.info("addRequestInterceptorFirst");
//            startMs.set(System.currentTimeMillis());
//          }
//        })
//                        .addExecInterceptorAfter(ChainElement.PROTOCOL.name(), "custom", (request, pr, scope, chain, c) -> {
//                                      startMs.set(System.currentTimeMillis());
//
//                            chain.proceed(request, pr, scope, c);
//                        })
        .build();

    final StringBody inputBody = new StringBody(
        "{\"content_type\":\"text/plain;charset=iso8859-1\",\"data\":\"将插头插入车内点烟器插孔 将插头插入车内点烟器插孔。冒昧通信，以期待与您建立业务关系。希望此邮件能够帮助您的收款要求。\",\"content_encoding\":\"8bit\"}",
        ContentType.APPLICATION_JSON);
    //    final StringBody inputBody = new StringBody(longVocsInputJson, ContentType.APPLICATION_JSON);
    final StringBody parametersBody = new StringBody(
        "{\"voice\":\"lilian-ml\",\"sample_rate\":\"22050\",\"output_type\":\"10\",\"session_id\":\"jmeter-apachehc-vocs-client\","
            + "\"language\":\"cmn-cn\",\"log_event\":\"1\",\"stream_mode\":\"2\"}",
        ContentType.APPLICATION_JSON);

    final HttpEntity reqEntity = MultipartEntityBuilder.create()
        .addPart("Input", inputBody)
        .addPart("Parameters", parametersBody)
        .build();
    String contentType = reqEntity.getContentType();

    client.start();
    final HttpHost target;
    if (host.contains(":")) {
      String[] ip = host.split(":");
      target = new HttpHost(ip[0], Integer.valueOf(ip[1]));
    } else {
      target = new HttpHost(host);
    }

    ByteArrayOutputStream baos = new ByteArrayOutputStream();
    reqEntity.writeTo(baos);
    byte[] bytes = baos.toByteArray();
    TimeBasicAsyncEntityProducer multiPartProducer = new TimeBasicAsyncEntityProducer(
        bytes,startMs
    );
    //    SimpleBody simpleBody=new SimpleBody(null,json,ContentType.APPLICATION_JSON);
    final SimpleHttpRequest request = SimpleRequestBuilder
        .post()
        .setHttpHost(target)
        .setScheme(schema())
        .setPath(vocsPath())
        .setHeader("Content-Type", contentType)
        //                        .addHeader("Content-Length",String.valueOf(bytes.length))

        //        .setPath("/get")
        //        .setPath("/post")
        //        .removeHeaders("Transfer-Encoding")
        //        .setBody(bytes, ContentType.MULTIPART_FORM_DATA)
        //        .setBody(json,ContentType.APPLICATION_JSON)
        //        .removeHeaders("transfer-encoding")
        .build();

    System.out.println(request.getHeader("Content-Length"));
    System.out.println("Executing request " + request);

    final CountDownLatch latch = new CountDownLatch(1);

    MultipartContext context = new MultipartContext("multipart/form-data; boundary=Nuance_VoCS_0", -1, "ISO-8859-1");

    NioMultipartParserListener listener = new NioMultipartParserListener() {

      @Override
      public void onPartFinished(final StreamStorage partBodyStreamStorage, final Map<String, List<String>> headersFromPart) {
        //      log.info("onPartFinished --->");
        boolean isAudio = headersFromPart.get("content-disposition").stream().anyMatch(x -> x.contains("Audio"));
        if (isAudio) {
          log.info("-audio {}", partBodyStreamStorage.getBytesCount());
          //        try {
          //          Files.write(audioPath, partBodyStreamStorage.getInputStream().readAllBytes(), StandardOpenOption.APPEND);
          //        } catch (IOException e) {
          //          e.printStackTrace();
          //        }

        } else {
          //        log.info(VocsUtils.convertStreamToString(partBodyStreamStorage.getInputStream()));
          log.info("-non-audio-part");
        }
        //      log.info("<---onPartFinished");

        // The parser completed parsing the part.
        // The parsed headers are available in the headersFromPart map
        // The part body can be read from the partBodyStreamStorage which provides an InputStream
      }

      @Override
      public void onNestedPartStarted(final Map<String, List<String>> headersFromParentPart) {
        // The parser identified that the current part contains a nested multipart body
        // The headers are provided in the headersFromParentPart
        // Like for the level 0 parts, the parser will notify the sub parts completion via the methods onPartFinished(...) and onFormFieldPartFinished(...)
        // When the nested multipart body is finished, the parser will call onNestedPartFinished()
      }

      @Override
      public void onNestedPartFinished() {
        // Called when a nested multipart body has been parsed
      }

      @Override
      public void onAllPartsFinished() {
        // Called when the multipart processing finished (encountered the close boundary). No more parts are available!
      }

      @Override
      public void onError(String message, Throwable cause) {
        // An error happened during the parsing.
        // At this point the parser is in an error state and it cannot process any data anymore.
      }
    };
    NioMultipartParser parser = Multipart.multipart(context)
        .usePartBodyStreamStorageFactory(countFactory)
        .forNIO(listener);

    client.execute(
        new BasicRequestProducer(request,
            //            basicAsyncEntityProducer
            multiPartProducer
        ),
        new VocsMultiPartResponseConsumer(latch, startMs, parser), null);
    try {
      latch.await(30, TimeUnit.SECONDS);
    } catch (InterruptedException e) {
      e.printStackTrace();

    }

  }
}

